package com.example.messagequeue.mqserver;

import com.example.messagequeue.commom.Consumer;
import com.example.messagequeue.commom.MqException;
import com.example.messagequeue.mqserver.core.*;
import com.example.messagequeue.mqserver.datacenter.DisDataCenter;
import com.example.messagequeue.mqserver.datacenter.MemoryDataCenter;
import jakarta.websocket.RemoteEndpoint;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/*
    通过这个类，来表示虚拟主机
    每个虚拟主机下面都管理者自己的交换机，队列，绑定，消息数据
    同时提供api提供给上层使用
 */
public class VirtualHost {
    private DisDataCenter disDataCenter = new DisDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();

    private Router router = new Router();
    private String virtualHostName;

    private ConsumerManager consumerManager = new ConsumerManager(this);

    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public DisDataCenter getDisDataCenter() {
        return disDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public VirtualHost(String virtualHostName){
        this.virtualHostName = virtualHostName;
        //对于MemoryDataCenter来说，不需要额外的初始化，只需要new出来就行
        //对于DisdataCenter来说,则需要进行初始化来进行建库建表和初始数据的设定
        disDataCenter.init();
        //另外还需要针对硬盘的数据，恢复到硬盘中
        try {
            memoryDataCenter.recovery(disDataCenter);
        } catch (IOException | ClassNotFoundException | MqException e) {
            e.printStackTrace();
            System.out.println("[virtualHost] 回复内存数据失败");
        }
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> argument){
        //把交换机的名字加入虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null){
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(argument);
                if (durable){
                    disDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchang(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName = " + exchangeName);
            }
            return true;
            //上述逻辑，先写硬盘，后写内存，目的就是因为硬盘更容易写失败，如果硬盘写失败了，内存就不写了
            //要是先写内存，内存写成功了，硬盘写失败了，还需要把内存的数据给在删除掉，就比较麻烦
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败 exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName){
        exchangeName =virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange == null){
                    throw new Exception("[VirtualHost] 交换机不存在无法删除");
                }
                if (existsExchange.isDurable()){
                    disDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功 exchangeName = " + exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除失败 exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments){
//        queueName = virtualHostName + queueName;
//        try {
//            synchronized (queueLocker){
//                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
//                if (existsQueue != null){
//                    System.out.println("[VirtualHost] 队列已经存在 queueName = " + queueName);
//                    return true;
//                }
//                MSGQueue queue = new MSGQueue();
//                queue.setName(queueName);
//                queue.setExclusive(exclusive);
//                queue.setDurable(durable);
//                queue.setAutoDelete(autoDelete);
//                queue.setArguments(argument);
//                if (durable){
//                    disDataCenter.insertQueue(queue);
//                }
//                memoryDataCenter.insertQueue(queue);
//                System.out.println("[VirtualHost] 队列创建完成! queueName = " + queueName);
//            }
//            return true;
//            //上述逻辑，先写硬盘，后写内存，目的就是因为硬盘更容易写失败，如果硬盘写失败了，内存就不写了
//            //要是先写内存，内存写成功了，硬盘写失败了，还需要把内存的数据给在删除掉，就比较麻烦
//        }catch (Exception e){
//            System.out.println("[VirtualHost] 队列创建失败 queueName = " + queueName);
//            e.printStackTrace();
//            return false;
//        }
        // 把队列的名字, 给拼接上虚拟主机的名字.
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                // 2. 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 写硬盘
                if (durable) {
                    disDataCenter.insertQueue(queue);
                }
                // 4. 写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null){
                    throw new MqException("[VirtualHost] 队列不存在无法删除 queueName = " + queueName);
                }
                if (existsQueue.isDurable()){
                    disDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功 queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列删除失败 queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding existsbinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsbinding != null){
                        throw new MqException("[VirtualHost] 绑定已经存在 exchangeName = " + exchangeName +
                                ", queueName = " + queueName);
                    }
                    if (!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey = " + bindingKey);
                    }
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName = " + exchangeName);
                    }
                    if (queue == null){
                        throw new MqException("[VirtualHost] 队列不存在! queueName = " + queueName);
                    }
                    if (exchange.isDurable() && queue.isDurable()){
                        disDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功 exchangeName = " + exchangeName +
                            ", queueName = " + queueName);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败 exchangeName = " + exchangeName +
                    ", queueName = " + queueName);
            e.printStackTrace();
            return false;
        }

    }

    public boolean queueUnbind(String exchangeName, String queueName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null){
                        throw new MqException("[VirtualHost] 绑定不存在 exchangeName = " + exchangeName +
                                ", queueName = " + queueName);
                    }
                    //为避免在删除绑定前，将交换机或者队列给删除了
                    //无论绑定是否持久化，都尝试从硬盘删除一下，就算不存在，这个删除也无副作用
                    disDataCenter.deleteBinding(binding);

                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 绑定删除成功 exchangeName = " + exchangeName +
                            ", queueName = " + queueName);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定删除失败");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try {
            exchangeName = virtualHostName + exchangeName;
            //检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey 非法 routingKey = " + routingKey);
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null){
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName = " + exchangeName);
            }
            //判断交换机类型
            if (exchange.getType() == ExchangeType.DIRECT){
                //按照直接交换机的方式转发消息
                //以routingKey作为队列的名字，直接把消息写入到指定队列中
                //此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                //查找相应的队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName = " + queueName);
                }
                sendMessage(queue, message);
            }else {
                //按照fanout和topic的方式转发
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBinding(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingsMap.entrySet()){
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null){
                        //此处就不抛出异常，可能有多个这样的队列
                        //希望不要因为一个队列的失败，影响到其他队列的消息的传输
                        System.out.println("[VirtualHost] basicPublish 发送消息时，发现队列不存在！queueName = " + queue.getName());
                        continue;
                    }
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    //判定这个消息是否能转发给该队列
                    //如果是fanout，所有绑定的队列都要转发
                    //如果是topic，还需要判定bindingKey和routingKey是否一致
                    if (!router.route(exchange.getType(), binding, message)){
                        continue;
                    }
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 发送消息失败");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        //deliverMode 为1表示不持久化，为2表示持久化
        if (deliverMode == 2){
            disDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue, message);
        //
        consumerManager.notifyConsume(queue.getName());
    }

    //订阅消息
    //添加一个队列的订阅者，当队列收到消息之后，就要把消息推送到对应的订阅者
    //consumerTag：消费者的身份标识
    //autoAck：消息被消费完成后，应答的方式。ture为自动应答，false为手动应答
    //consumer：是一个回调函数，此处类型设定为函数式接口，这样后续用basicConsume并且传实参时，就可以用lambda表达式
    public boolean basicConsume(String consumerTag, String queueName,boolean autoAck, Consumer consumer){
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName = " + queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicConsume 失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //对未确认消息进行删除

    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try {
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在! messageId = " + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null){
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName = " + queueName);
            }
            if (message.getDeliverMode() == 2){
                disDataCenter.deleteMessage(queue,message);
            }
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            memoryDataCenter.removeMessage(messageId);
            System.out.println("[VirtualHost] basic成功! 消息被成功确认！queueName = " + queueName + ", messageId = " + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basic失败! 消息确认失败！queueName = " + queueName + ", messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }

}
